package com.eascs.web.o2o.mq;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.TimeUnit;

import javax.inject.Inject;

import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Autowired;

import com.alibaba.fastjson.JSON;
import com.eascs.app.log.Logger;
import com.eascs.app.log.LoggerFactory;
import com.eascs.notify.constant.ModelResult;
import com.eascs.web.o2o.mq.entity.LogEntity;
import com.eascs.web.o2o.mq.entity.RepayMentInfoEntity;
import com.eascs.web.o2o.mq.service.TransService;
import com.eascs.web.o2o.mq.service.impl.LogService;
import com.rabbitmq.client.AMQP;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.DefaultConsumer;
import com.rabbitmq.client.Envelope;

public class LoanConsumer<T> extends EndPoint implements InitializingBean{
	
	private static final Logger logger = LoggerFactory.getLogger(LoanConsumer.class);
	private Map<String, TransService<T>> serviceMap = new HashMap<String, TransService<T>>();
	
	private Channel  channel;
	
	@Inject
	private LogService  logService;
	
	
	@Autowired
	public void setServiceMap(List<TransService<T>> interfaceList) {//spring管理实现注入接口
		for (TransService<T> service : interfaceList) {
			serviceMap.put(service.getServiceName(), service);
		}
	}
	
	@Override
	public void afterPropertiesSet() throws Exception {//spring管理项目启动执行
		executorService.execute(new Runnable() {
			@Override
			public void run() {//创建独立线程去执行，避免mq资源链接不上导致容器启动失败的异常
				consumer();
			}
		});
	}
	
	public void consumer(){
		channel =  buildChannel();
		try {
			// 每次消费的消息数
			channel.basicQos(1);
			// 消息消费完成确认
			channel.basicConsume(INTERFACE_TASK_QUEUE_NAME, false, new LoanConsumerHandler(channel));
			logger.info("rabbit mq consumer start success");
		} catch (IOException e) {
			logger.error("rabbit mq consumer start err", e);
		}
		
		
	}
	/**
	 * comsumer handler
	 */
	class LoanConsumerHandler extends DefaultConsumer{

		public LoanConsumerHandler(Channel channel) {
			super(channel);
		}
		
		@SuppressWarnings("unchecked")
		@Override
		public void handleDelivery(String consumerTag,Envelope envelope,AMQP.BasicProperties properties,byte[] body) throws IOException{
			
			String consumerMessage =new String(body, "UTF-8");
			logger.info("comsumer received mq msg :[{}]", consumerMessage);
			ModelResult<Boolean> result = new ModelResult<Boolean>();
			try {
				Message message = JSON.parseObject(consumerMessage, Message.class);
				if(message.getTransType().equals("RepayMent")){
					RepayMentInfoEntity repayMentInfoEntity = JSON.parseObject(message.getBody(), RepayMentInfoEntity.class);//获取消费对象(service)
					if(repayMentInfoEntity!=null){
						LogEntity logEntity= logService.findEntity(repayMentInfoEntity.getLoanappyNo(),"RepayMent",0);//判断消息是否被消费
						if(logEntity ==  null){
							TransService<RepayMentInfoEntity>	transService = (TransService<RepayMentInfoEntity>) serviceMap.get(message.getTransType());
							result = transService.exec(repayMentInfoEntity);//消费消息
							TimeUnit.SECONDS.sleep(10);
						}else{
							result.withError("RepayMent重复消息丢掉", "RepayMent重复消息丢掉");
						}
					}else{
						result.withError("RepayMent消息为空", "RepayMent消息为空");
					}
					
				}
			} catch (Exception e) {
				logger.error("consume msg error", e);
			}finally {
				logger.info("consume msg result:[{}]", JSON.toJSONString(result));
				if (result.isSuccess()) {
					// 确认消息
					channel.basicAck(envelope.getDeliveryTag(), false);
				} else {
					// 丢掉消息
					channel.basicNack(envelope.getDeliveryTag(), false, false);
				}
			}
			
		}
	}
	
	

}
